package service

import java.sql.{Connection, DriverManager}
import java.util.Properties

import dao.EmploymentStatusDAO
import org.apache.spark.sql.SparkSession

object EmploymentStatusService {

    def dataProcess(spark:SparkSession,url2:String,table1:String,props:Properties,
                    url4:String,props4:Properties)={
        //元数据
        EmploymentStatusDAO.getDate(spark,url2,table1,props).createOrReplaceTempView("t1")


        spark.sql(
            """
              |
              |select
              |	name,
              |	'企业招聘' content,
              |	edu_time
              |from t1
              |where enterprise_recruitment is not null
              |
              |union all
              |select
              |	name,
              |	'自主创业' content,
              |	edu_time
              |from t1
              |where autonomy_work is not null
              |
              |union all
              |select
              |	name,
              |	'单位就业' content,
              |	edu_time
              |from t1
              |where company_work is not null
              |
              |union all
              |select
              |	name,
              |	'特殊岗位就业' content,
              |	edu_time
              |from t1
              |where special_position_work is not null
              |
              |union all
              |select
              |	name,
              |	'未就业' content,
              |	edu_time
              |from t1
              |where unemploy_type is not null
              |
              |union all
              |select
              |	name,
              |	'其他' content,
              |	edu_time
              |from t1
              |where work_other is not null  --人，就业情况，时间
              |
              |
            """.stripMargin).createOrReplaceTempView("t2")

        spark.sql("select name, content, date_format(edu_time,'yyyy') as df_year from t2")
          .createOrReplaceTempView("t3")


        val df = spark.sql(
            """

            select
                count(1) as num,--应届毕业生就业情况
                concat("应届",content) as content,
             '毕业生就业情况' as name
            from (
                select
                    name,
                    content,
                    df_year
                from t3
                where df_year=date_format(current_date(),'yyyy')  ---今年毕业生
                )a
            group by content

            union all
            select
                count(1) as num,--往届毕业生就业情况
                concat("往届",content) as content,
             '毕业生就业情况' as name
            from (
                select
                    name,
                    content,
                    df_year
                from t3
                where df_year != date_format(current_date(),'yyyy')  ---往年毕业生
                )a
            group by content

            """.stripMargin)

       // df.show()

        var connection: Connection = null
        df.foreachPartition(iter => {
            try {
                classOf[com.mysql.jdbc.Driver]
                connection = DriverManager.getConnection(url4, props4)
                while (iter.hasNext) {
                    val row = iter.next() //每行数据 num,content,name
                    val num = row.get(0)
                    val content = row.getAs[String]("content")
                    val name = row.get(2).toString

                    val sql = s"insert into test.big_screen_result_copy1(num,content,name) values('$num','$content','$name')"
                    connection.createStatement().executeUpdate(sql)
                }
            }   catch {
                case e: Exception => println(e.printStackTrace())
            } finally {
                connection.close()
            }
        })

    }


}
